package com.fang.mmy.items.mq;

import lombok.val;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;

/**
 * @ClassName LoginMQConsumer
 * @Author kebukeyi
 * @Date 2022/8/12 11:38
 * @Description
 * @Version 1.0.0
 */
@Component
//@RocketMQMessageListener(consumerGroup = "Con_Group_One", topic = "user_login", consumeMode = ConsumeMode.CONCURRENTLY)
public class LoginMQConsumer implements MessageListenerConcurrently {


    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        try {
            for (MessageExt message : list) {
                System.out.println("开始处理订单数据，准备增加积分....");
                //这里虽然获得消息了  也提交了 offset 给mq了 ;但是系统突然宕机了 那就还是还是消费失败了 下发红包失败
                val str = new String(message.getBody(), StandardCharsets.UTF_8);
                System.out.println(str);
                //但是这里我们要警惕一点，就是我们不能在代码中对消息进行异步的处理，如下错误的示范，我们开启了一个子线程去处理这批消息，
                // 然后启动线程之后，就直接返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS状态了
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            System.out.printf("处理消费者数据发生异常 {}", e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}
